package com.yunhan.redis;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.yunhan.utils.ExecutorsUtils;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.redisson.RedissonMultiLock;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.MDC;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.StringRedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


@Slf4j
@Setter
public class RedisCacheImpl implements RedisCache {

    private final int EXPIRY_TIME = 5;

    private StringRedisTemplate redisTemplate;

    private RedissonClient redissonClient;

    /**
     * 从双缓存获取值
     * @param key1 缓存key
     * @param key2 缓存key
     * @return
     */
    @Override
    public <T> T getDoubleCache(String key1, String key2, Class<T> clazz) {
        //从缓存1取值
        Object result = getSerializ(key1, clazz);

        if (result == null) {
            //缓存1为命中，则从缓存2
            result = getSerializ(key2, clazz);
        }
        return clazz.cast(result);
    }


    /**
     * 双缓存需要序列化，击穿1缓存异步构建
     * 缓存设置执行数据代理接口
     * key1（默认缓存2分钟）
     * key2（默认缓存1小时）
     *
     * @param key1          缓存key
     * @param cacheConsumer 业务执行
     * @param <T>
     * @return
     */
    @Override
    public <T> T getIfPresentDouble(String key1, String key2, CacheConsumer cacheConsumer, Class<T> clazz) {
        //缓存1
        Object result = getSerializ(key1, clazz);
        if (result == null) {
            //缓存2
            result = getSerializ(key2, clazz);
            if (result == null) {
                result = cacheConsumer.execute();
                buildCache(key1, CacheTime.Minute.MINUTE_2, key2, CacheTime.Hour.HOUR_1, result, TimeUnit.SECONDS);
            } else {
                //异步构建缓存
                Map<String, String> contextMap = MDC.getCopyOfContextMap();
                ExecutorsUtils.getExecutorPool().execute(() -> {
                    try {
                        MDC.setContextMap(contextMap);
                        buildCache(key1, CacheTime.Minute.MINUTE_2, key2, CacheTime.Hour.HOUR_1, cacheConsumer.execute(), TimeUnit.SECONDS);
                    } finally {
                        MDC.clear();
                    }
                });
            }
        }
        return clazz.cast(result);
    }


    /**
     * 双缓存需要序列化，击穿1缓存异步构建
     * 缓存设置执行数据代理接口
     *
     * @param key1          缓存Key1
     * @param expireTime1   key1过期时间
     * @param key2          缓存Key2
     * @param expireTime2   key2过期时间
     * @param cacheConsumer 执行逻辑
     * @param unit          时间单位
     * @param clazz         返回结果类型
     * @return
     */
    @Override
    public <T> T getIfPresentDouble(String key1, Long expireTime1, String key2, Long expireTime2, CacheConsumer cacheConsumer, TimeUnit unit, Class<T> clazz) {
        Object result = getSerializ(key1, clazz);
        if (result == null) {
            //缓存2
            result = getSerializ(key2, clazz);
            if (result == null) {
                result = cacheConsumer.execute();
                buildCache(key1, expireTime1, key2, expireTime2, result, unit);
            } else {
                //异步构建缓存
                Map<String, String> contextMap = MDC.getCopyOfContextMap();
                ExecutorsUtils.getExecutorPool().execute(() -> {
                    try {
                        MDC.setContextMap(contextMap);

                        buildCache(key1, expireTime1, key2, expireTime2, cacheConsumer.execute(), unit);
                    } finally {
                        MDC.clear();
                    }
                });
            }
        }
        return clazz.cast(result);
    }


    /**
     * 双缓存需要序列化，击穿1缓存异步构建
     * 缓存设置执行数据代理接口
     * key1（默认缓存2分钟）
     * key2（默认缓存1小时）
     *
     * @param key1          缓存key
     * @param cacheConsumer 业务执行
     * @param <T>
     * @return
     */
    @Override
    public <T> T getIfPresentDoubleArray(String key1, String key2, CacheConsumer cacheConsumer, Class<T> clazz) {
        //缓存1
        Object result = getSerializArray(key1, clazz);
        if (result == null) {
            //缓存2
            result = getSerializArray(key2, clazz);
            if (result == null) {
                result = cacheConsumer.execute();
                buildCache(key1, CacheTime.Minute.MINUTE_2, key2, CacheTime.Hour.HOUR_1, result, TimeUnit.SECONDS);
            } else {
                //异步构建缓存
                Map<String, String> contextMap = MDC.getCopyOfContextMap();
                ExecutorsUtils.getExecutorPool().execute(() -> {
                    try {
                        MDC.setContextMap(contextMap);
                        buildCache(key1, CacheTime.Minute.MINUTE_2, key2, CacheTime.Hour.HOUR_1, cacheConsumer.execute(), TimeUnit.SECONDS);
                    } finally {
                        MDC.clear();
                    }
                });
            }
        }
        return clazz.cast(result);
    }

    /**
     * 双缓存需要序列化，击穿1缓存异步构建
     * 缓存设置执行数据代理接口
     *
     * @param key1          缓存Key1
     * @param expireTime1   key1过期时间
     * @param key2          缓存Key2
     * @param expireTime2   key2过期时间
     * @param cacheConsumer 执行逻辑
     * @param unit          时间单位
     * @param clazz         返回结果类型
     * @return
     */
    @Override
    public <T> T getIfPresentDoubleArray(String key1, Long expireTime1, String key2, Long expireTime2, CacheConsumer cacheConsumer, TimeUnit unit, Class<T> clazz) {
        Object result = getSerializArray(key1, clazz);
        if (result == null) {
            //缓存2
            result = getSerializArray(key2, clazz);
            if (result == null) {
                result = cacheConsumer.execute();
                buildCache(key1, expireTime1, key2, expireTime2, result, unit);
            } else {
                //异步构建缓存
                Map<String, String> contextMap = MDC.getCopyOfContextMap();
                ExecutorsUtils.getExecutorPool().execute(() -> {
                    try {
                        MDC.setContextMap(contextMap);
                        buildCache(key1, expireTime1, key2, expireTime2, cacheConsumer.execute(), unit);
                    } finally {
                        MDC.clear();
                    }
                });
            }
        }
        return clazz.cast(result);
    }


    /**
     * 单缓存数据访问设置接口（默认缓存5分钟）
     * 当缓存未命中时执行自定义业务，将业务结果返回和设设置到redis中
     * @param key           缓存key
     * @param cacheConsumer 缓存未命中时要执行的业务
     */
    @Override
    public <T> T getIfPresent(String key, CacheConsumer cacheConsumer, Class<T> clazz) {
        Object result = getSerializ(key, clazz);
        if (result == null) {
            result = cacheConsumer.execute();
            if (result == null) {
                return null;
            }
            if (result instanceof String) {
                set(key, (String) result);
            } else {
                set(key, result);
            }
        }
        return clazz.cast(result);
    }


    /**
     * 缓存设置执行数据代理接口
     *
     * @param key           缓存key
     * @param cacheConsumer 业务执行
     * @param expireTime    过期时间
     * @param unit          时间单位
     * @param <T>
     * @return
     */
    @Override
    public <T> T getIfPresent(String key, CacheConsumer cacheConsumer, Long expireTime, TimeUnit unit, Class<T> clazz) {
        Object result = getSerializ(key, clazz);
        if (result == null) {
            result = cacheConsumer.execute();
            if (result == null) {
                return null;
            }
            if (result instanceof String) {
                if (expireTime == null || unit == null) {
                    set(key, (String) result);
                } else {
                    set(key, (String) result, expireTime, unit);
                }
            } else {
                if (expireTime == null || unit == null) {
                    set(key, result);
                } else {
                    set(key, result, expireTime, unit);
                }

            }
        }
        return clazz.cast(result);
    }

    /**
     *
     * 单缓存数据访问设置接口，返回值类型为集合
     * 缓存设置执行数据代理接口（默认缓存5分钟）
     * @param key           缓存key
     * @param cacheConsumer 业务执行
     */
    @Override
    public <T> T getIfPresentArray(String key, CacheConsumer cacheConsumer, Class<T> clazz) {
        Object result = getSerializArray(key, clazz);

        if (result == null) {
            result = cacheConsumer.execute();
            if (result == null) {
                return null;
            }
            if (result instanceof String) {
                set(key, (String) result);
            } else {
                set(key, result);
            }
        }
        return clazz.cast(result);
    }

    /**
     * 缓存设置执行数据代理接口阻塞锁
     *
     * @param key           缓存key
     * @param cacheConsumer 业务执行
     * @param expireTime    过期时间
     * @param unit          时间单位
     * @param <T>
     * @return
     */
    @Override
    public <T> T getIfPresentForLock(String key, String lockKey, Long lockExpireTime, TimeUnit lockUnit, CacheConsumer cacheConsumer, Long expireTime, TimeUnit unit, Class<T> clazz) {
        Object result = getSerializ(key, clazz);
        if (result == null) {
            //阻塞锁
            lock(lockKey, lockUnit, lockExpireTime);
            try {
                //二次兜底
                result = getSerializ(key, clazz);
                if (result == null) {
                    result = cacheConsumer.execute();
                    if (result == null) {
                        return null;
                    }
                    if (result instanceof String) {
                        if (expireTime == null || unit == null) {
                            set(key, (String) result);
                        } else {
                            set(key, (String) result, expireTime, unit);
                        }
                    } else {
                        if (expireTime == null || unit == null) {
                            set(key, result);
                        } else {
                            set(key, result, expireTime, unit);
                        }

                    }
                }
            } finally {
                unlock(lockKey);
            }
        }
        return clazz.cast(result);
    }

    /**
     * 缓存设置执行数据代理接口阻塞锁
     *
     * @param key           缓存key
     * @param cacheConsumer 业务执行
     * @param expireTime    过期时间
     * @param unit          时间单位
     * @param <T>
     * @return
     */
    @Override
    public <T> T getIfPresentForTryLock(String key, String lockKey, Long lockExpireTime, TimeUnit lockUnit, CacheConsumer cacheConsumer, Long expireTime, TimeUnit unit, Class<T> clazz) {
        Object result = getSerializ(key, clazz);
        if (result == null) {
            //幂等锁
            if (!tryLock(lockKey, lockUnit, lockExpireTime)) {
                return null;
            }
            try {
                result = cacheConsumer.execute();
                if (result == null) {
                    return null;
                }
                if (result instanceof String) {
                    if (expireTime == null || unit == null) {
                        set(key, (String) result);
                    } else {
                        set(key, (String) result, expireTime, unit);
                    }
                } else {
                    if (expireTime == null || unit == null) {
                        set(key, result);
                    } else {
                        set(key, result, expireTime, unit);
                    }

                }
            } finally {
                unlock(lockKey);
            }
        }
        return clazz.cast(result);
    }






    /**
     * 双缓存需要序列化，击穿1缓存异步构建
     * 缓存设置执行数据代理接口（锁）
     *
     * @param key1           缓存Key1
     * @param expireTime1    key1过期时间
     * @param key2           缓存Key2
     * @param expireTime2    key2过期时间
     * @param lockKey        锁
     * @param lockExpireTime 锁过期时间
     * @param cacheConsumer  执行逻辑
     * @param unit           时间单位
     * @param clazz          返回结果类型
     * @return
     */
    @Override
    public <T> T getIfPresentDoubleForLock(String key1, Long expireTime1, String key2, Long expireTime2, String lockKey, Long lockExpireTime, CacheConsumer cacheConsumer, TimeUnit unit, Class<T> clazz) {
        Object result = getSerializ(key1, clazz);
        if (result != null) {
            return clazz.cast(result);
        }
        //缓存2
        result = getSerializ(key2, clazz);

        if (result == null) {
            //阻塞锁
            lock(lockKey, unit, lockExpireTime);
            try {
                //兜底再次查询缓存2
                result = getSerializ(key2, clazz);
                if (result == null) {
                    result = cacheConsumer.execute();
                    buildCache(key1, expireTime1, key2, expireTime2, result, unit);
                }
            } finally {
                unlock(lockKey);
            }

        } else {
            //异步构建缓存
            Map<String, String> contextMap = MDC.getCopyOfContextMap();
            ExecutorsUtils.getExecutorPool().execute(() -> {
                //幂等锁
                if (!tryLock(lockKey, unit, lockExpireTime)) {
                    return;
                }
                try {
                    MDC.setContextMap(contextMap);
                    buildCache(key1, expireTime1, key2, expireTime2, cacheConsumer.execute(), unit);
                } finally {
                    unlock(lockKey);
                    MDC.clear();
                }
            });
        }
        return clazz.cast(result);
    }



    /**
     * 设置对象缓存（默认5分钟）
     *
     * @param key key
     * @param obj 对象
     */
    @Override
    public void set(String key, Object obj) {
        Assert.notNull(obj, "value cannot be null");
        redisTemplate.opsForValue().set(key, JSON.toJSONString(obj), EXPIRY_TIME, TimeUnit.MINUTES);
    }

    /**
     * 设置对象缓存（自定义过期时间）
     *
     * @param key    key
     * @param obj    对象
     * @param expiry 过期时间
     * @param unit   时间单位
     */
    @Override
    public void set(String key, Object obj, long expiry, TimeUnit unit) {
        Assert.notNull(obj, "value cannot be null");
        redisTemplate.opsForValue().set(key, JSON.toJSONString(obj), expiry, unit);
    }



    /**
     * 设置字符串缓存
     * @param key key
     * @param str 字符串
     */
    @Override
    public void set(String key, String str) {
        Assert.notNull(str, "value cannot be null");
        redisTemplate.opsForValue().set(key, str);
    }

    @Override
    public void setInt(String key, Integer val) {
        Assert.notNull(val, "value cannot be null");
        redisTemplate.opsForValue().set(key, val.toString());
    }

    /**
     * 设置字符串缓存（指定日期到期）
     * @param key key
     * @param str 字符串
     * @param date 到期日期
     */
    @Override
    public void set(String key, String str, Date date) {
        Assert.notNull(str, "value cannot be null");
        redisTemplate.opsForValue().set(key, str);
        redisTemplate.expireAt(key, date);
    }



    /**
     * 设置字符串缓存
     *
     * @param key     key
     * @param content 值
     * @param expiry  过期时间
     * @param unit    时间单位
     */
    @Override
    public void set(String key, String content, long expiry, TimeUnit unit) {
        Assert.notNull(content, "value cannot be null");
        redisTemplate.opsForValue().set(key, content, expiry, unit);
    }

    /**
     * 把数据设置到hash中（自定义过期时间）
     * @param key  key
     * @param hKey hash的key
     * @param val  值
     * @param num  过期时间
     */
    @Override
    public void hSet(String key, String hKey, String val, long num, TimeUnit tunit) {
        Assert.notNull(val, "value cannot be null");
        redisTemplate.opsForHash().put(key, hKey, val);
        redisTemplate.expire(key, num, tunit);
    }

    /**
     * 把数据设置到hash中（指定日期过期）
     * @param key  key
     * @param hKey hash的key
     * @param val  值
     * @param date 过期时间
     */
    @Override
    public void hSet(String key, String hKey, String val, Date date) {
        Assert.notNull(val, "value cannot be null");
        redisTemplate.opsForHash().put(key, hKey, val);
        redisTemplate.expireAt(key, date);
    }

    /**
     * set集合添加多个字符串元素
     * @param key
     * @param vals
     */
    @Override
    public Long sAdd(String key, String[] vals) {
        Assert.notNull(vals, "value cannot be null");
        Long add = redisTemplate.opsForSet().add(key, vals);
        return add;

    }

    @Override
    public Long sAdd(String key, String vals) {
        Assert.notNull(vals, "value cannot be null");
        return redisTemplate.opsForSet().add(key, vals);
    }


    @Override
    public Long szAdd(String key, Set<ZSetOperations.TypedTuple<String>> vals) {
        Assert.notNull(vals, "value cannot be null");
        Long add = redisTemplate.opsForZSet().add(key, vals);
        return add;

    }

    @Override
    public boolean szAdd(String key, String val, Double score) {
        return Boolean.TRUE.equals(redisTemplate.opsForZSet().add(key, val, score));
    }

    @Override
    public String szPop(String key) {
        //TODO 栈溢出
        String value = Objects.requireNonNull(redisTemplate.opsForZSet().popMin(key)).getValue();
        return value;
    }

    /**
     * 设置过期时间
     * @param key
     * @param unit
     * @param timeout
     */
    @Override
    public long setTtl(String key, TimeUnit unit, long timeout) {
        redisTemplate.expire(key, timeout, unit);
        return getExpire(key);
    }





    /**
     * 获取缓存
     * @param key key
     * @return 返回值字符串
     */
    @Override
    public String get(String key) {
        return redisTemplate.opsForValue().get(key);
    }


    /**
     * 获取缓存
     * @param key    key
     * @param clazz 类
     * @return 返回值类型自定义
     */
    @Override
    public <T> T get(String key, Class<T> clazz) {
        Object result = getSerializ(key, clazz);
        if (null == result)
            return null;
        return clazz.cast(result);
    }

    /**
     * 获取缓存
     * @param key    key
     * @param clazz 类
     * @return 返回值类型为List集合
     */
    @Override
    public <T> List<T> gets(String key, Class<T> clazz) {
        String result = get(key);
        if (!StringUtils.isEmpty(result)) {
            return JSON.parseArray(result, clazz);
        }
        return null;
    }

    /**
     * 获取hash的key
     * @param key  key
     * @param hKey hash的key
     * @return 返回值
     */
    @Override
    public Object hGet(String key, String hKey) {
        return redisTemplate.opsForHash().get(key, hKey);
    }

    /**
     * 获取整个hash
     * @param key key
     * @return 返回值
     */
    @Override
    public Map<Object, Object> hGet(String key) {
        return redisTemplate.opsForHash().entries(key);
    }

    /**
     * 获取剩余过期时间
     * @param key key
     */
    @Override
    public Long getExpire(String key) {
        return redisTemplate.getExpire(key);
    }

    /**
     * 从有序集合中获取指定分数范围内的元素
     *
     * @param key       缓存key
     * @param minScore  最小分数
     * @param maxScore  最大分数
     * @param offset    偏移量
     * @param count     返回数量
     * @return 返回的元素集合
     */
    @Override
    public Set<ZSetOperations.TypedTuple<String>> rangeByScoreWithScores(String key, double minScore, double maxScore, long offset, long count) {
        return redisTemplate.opsForZSet().rangeByScoreWithScores(key, minScore, maxScore, offset, count);

    }

    /**
     * 删除key
     * @param key key
     */
    @Override
    public void remove(String key) {
        redisTemplate.delete(key);
    }

    /**
     * 批量删除key
     * @param keys key
     */
    @Override
    public void remove(List<String> keys) {
        redisTemplate.delete(keys);
    }

    /**
     * 删除hash中的key
     * @param key  key redis的key
     * @param hKey hkey hash中的key
     */
    @Override
    public void hRomve(String key, String hKey) {
        redisTemplate.opsForHash().delete(key, hKey);
    }

    /**
     * 批量删除hash中的key
     * @param key   key
     * @param hKeys hkey
     */
    @Override
    public void hRomves(String key, String... hKeys) {
        redisTemplate.opsForHash().delete(key, hKeys);
    }

    /**
     * 删除set集合中指定元素
     * @param key set集合的key
     * @param obj 要删除的集合元素
     */
    @Override
    public long sRemove(String key, Object obj) {
        Assert.notNull(obj, "value cannot be null");
        Long success = redisTemplate.opsForSet().remove(key, obj);
        return ObjectUtils.defaultIfNull(success, 0L);
    }

    @Override
    public long zRemove(String key, String value) {
        try {
            return Optional.ofNullable(
                    redisTemplate.execute(connection ->
                                    connection.zRem(
                                            key.getBytes(StandardCharsets.UTF_8),
                                            value.getBytes(StandardCharsets.UTF_8)
                                    ),
                            true)
            ).orElse(0L);
        } catch (RedisSystemException e) {
            log.error("Redis操作异常 key:{}", key, e);
            return 0L;
        }

    }

    /**
     * set集合随机移除一个元素，并返回
     * @param key
     * @param clazz
     * @return
     */
    @Override
    public <T> T sPop(String key, Class<T> clazz) {
        String pop = redisTemplate.opsForSet().pop(key);
        if (StringUtils.isEmpty(pop)) {
            return null;
        }
        return JSON.parseObject(pop, clazz);
    }

    /**
     * set集合随机移除一个字符串，并返回
     * @param key
     * @return
     */
    @Override
    public String sPopStr(String key) {
        return redisTemplate.opsForSet().pop(key);
    }

    @Override
    public List<String> sMembers(String key) {
        Set<String> members = redisTemplate.opsForSet().members(key);
        return new ArrayList<>(members);
    }

    @Override
    public <T>List<T> sGetAndDistinct(String key, Integer num, Class<T> _class) {
        Set<String> distinctRandomMembers = redisTemplate.opsForSet().distinctRandomMembers(key, num);
        return distinctRandomMembers.stream().filter(StringUtils::isNotEmpty).map(x -> JSONObject.parseObject(x, _class)).collect(Collectors.toList());
    }

    @Override
    public Integer size(String key) {
        return redisTemplate.opsForSet().size(key).intValue();
    }

    @Override
    public String sGet(String key) {
        return redisTemplate.opsForSet().randomMember(key);
    }

    /**
     * set集合随机移除多个元素，并返回
     * @param key
     * @param count
     * @return
     */
    @Override
    public <T> List<T> sPopList(String key, long count, Class<T> _class) {
        List<String> pop = redisTemplate.opsForSet().pop(key, count);
        return pop.stream().filter(StringUtils::isNotEmpty).map(x -> JSONObject.parseObject(x, _class)).collect(Collectors.toList());
    }







    /**
     * 判断key是否存在
     * @param key 参数
     * @return 返回值
     */
    @Override
    public Boolean exists(String key) {
        return redisTemplate.hasKey(key);
    }

    @Override
    public void publish(String channel, String message) {
        redisTemplate.convertAndSend(channel, message);
    }

    /**
     * set集合判断元素是否存在
     * @param key
     * @param value
     * @return
     */
    @Override
    public Boolean isMember(String key, Object value) {
        Assert.notNull(value, "value cannot be null");
        return redisTemplate.opsForSet().isMember(key, value);
    }



    @Override
    public Set<String> members(String key) {
        return redisTemplate.opsForSet().members(key);
    }



    /**
     * setIfAbsent：尝试向redis中设置key，key不存在则设置，存在则设置不设置
     * @param key  key
     * @param val  value
     * @param time 过期时间
     * @return 返回值
     */
    @Override
    public boolean setIfAbsent(String key, String val, long time, TimeUnit unit) {
        if (StringUtils.isEmpty(key)) {
            return false;
        }
        Assert.notNull(val, "value cannot be null");
        Boolean success = redisTemplate.opsForValue().setIfAbsent(key, val);
        if (success != null && success) {
            redisTemplate.expire(key, time, unit);
            return true;
        }

        return false;
    }

    /**
     * setIfAbsent：尝试向hash中设置hkey，hkey不存在则设置，存在则设置不设置
     * @param key  key
     * @param hkey  hkey
     * @param val  hvalue
     * @param time key的过期时间
     * @return 返回值
     */
    @Override
    public boolean hSetIfAbsent(String key, String hkey, String val, long time, TimeUnit unit) {
        if (StringUtils.isEmpty(key)) {
            return false;
        }
        Assert.notNull(val, "value cannot be null");
        Boolean success = redisTemplate.opsForHash().putIfAbsent(key, hkey, val);
        if (success) {
            redisTemplate.expire(key, time, unit);
        }
        return success;
    }

    /**
     * lock-锁hkey
     * @param key  key
     * @param hkey  hkey
     * @param val  值
     * @param time key的过期时间
     * @return 返回值
     */
    @Override
    public boolean hLock(String key, String hkey, String val, long time, TimeUnit unit) {
        return hSetIfAbsent(key, hkey, val, time, unit);
    }


    /**
     * 原子增长
     * @param key
     * @param value
     * @param time
     * @return
     */
    @Override
    public long incrBy(String key, long value, long time, TimeUnit unit) {
        Assert.notNull(value, "value cannot be null");
        long index = redisTemplate.opsForValue().increment(key, value);
        redisTemplate.expire(key, time, unit);
        return index;
    }

    @Override
    public Integer incrBy(String key, Integer value) {
        Assert.notNull(value, "value cannot be null");
        return  redisTemplate.opsForValue().increment(key, value).intValue();
    }

    @Override
    public Integer decrBy(String key, Integer value) {
        Assert.notNull(value, "value cannot be null");
        return Objects.requireNonNull(redisTemplate.opsForValue().decrement(key, value)).intValue();
    }

    /**
     * hash的原子增长
     * @param key  hash的key
     * @param hKey key
     * @param val  值
     * @return 返回值
     */
    @Override
    public long hIncrBy(String key, String hKey, long val, long time, TimeUnit unit) {
        Assert.notNull(val, "value cannot be null");
        long result = redisTemplate.opsForHash().increment(key, hKey, val);
        redisTemplate.expire(key, time, unit);
        return result;
    }



    /**
     * 获取当前原子变量值（不存在会默认设置0）
     *
     * @param key
     * @param time
     * @return
     */
    @Override
    public long currentIncrby(String key, long time, TimeUnit unit) {
        long index = redisTemplate.opsForValue().increment(key, 0);
        redisTemplate.expire(key, time, unit);
        return index;
    }


    /**
     * 获取当前Hash原子变量值（不存在会默认设置0）
     * @param key
     * @param time
     * @return
     */
    @Override
    public long currentHashIncrby(String key, String filed, long time, TimeUnit unit) {
        long index = redisTemplate.opsForHash().increment(key, filed, 0);
        redisTemplate.expire(key, time, unit);
        return index;
    }

    /**
     * 管道批量执行(仅支持同一类型)
     */
    @Override
    public <T> List<T> batchGets(List<String> keys, Class<T> clazz) {

        List<Object> result = redisTemplate.executePipelined((RedisCallback<String>) connection -> {
            StringRedisConnection opt = (StringRedisConnection) connection;
            keys.forEach(opt::get);
            return null;
        });

        return result.stream().filter(Objects::nonNull).map(x -> JSONObject.parseObject(x.toString(), clazz)).collect(Collectors.toList());
    }

    /**
     * 管道批量执行
     */
    @Override
    public List<Object> pipelinedSetMembers(String key, List<String[]> values, long timeOut, TimeUnit unit) {

        List<Object> result = redisTemplate.executePipelined((RedisCallback<String>) connection -> {
            StringRedisConnection opt = (StringRedisConnection) connection;
            values.forEach(value -> opt.sAdd(key, value));
            return null;
        });

        setTtl(key, unit, timeOut);
        return result;
    }

    /**
     * 管道批量执行(仅支持同一类型)
     */
    @Override
    public List<String> batchGets(List<String> keys) {

        List<Object> result = redisTemplate.executePipelined((RedisCallback<String>) connection -> {
            StringRedisConnection opt = (StringRedisConnection) connection;
            keys.forEach(opt::get);
            return null;
        });

        return result.stream().filter(Objects::nonNull).map(Object::toString).collect(Collectors.toList());
    }

    /**
     * 批量执行(仅支持同一类型)
     */
    @Override
    public <T> List<T> mGets(List<String> keys, Class<T> _class) {

        List<String> result = redisTemplate.opsForValue().multiGet(keys);

        return result.stream().filter(StringUtils::isNotEmpty).map(x -> JSONObject.parseObject(x, _class)).collect(Collectors.toList());
    }

    /**
     * 批量执行(仅支持同一类型)
     */
    @Override
    public List<String> mGets(List<String> keys) {
        return redisTemplate.opsForValue().multiGet(keys);
    }

    /**
     * 获取列表指定范围内的元素
     *
     * @param key
     * @param start 开始位置, 0是开始位置
     * @param end   结束位置, -1返回所有
     * @return
     */
    @Override
    public List<String> lRange(String key, long start, long end) {
        return redisTemplate.opsForList().range(key, start, end);
    }



    /**
     * 管道批量执行删除
     *
     * @param keys
     */
    @Override
    public void batchDel(List<String> keys) {
        if (CollectionUtils.isEmpty(keys)) {
            return;
        }
        redisTemplate.executePipelined((RedisCallback<String>) connection -> {
            StringRedisConnection opt = (StringRedisConnection) connection;
            opt.del(ArrayUtils.toStringArray(keys.toArray()));
            return null;
        });
    }

    /**
     * 阻塞直到加锁成功
     * @param lockKey
     * @param unit
     * @param timeout
     */
    @Override
    public RLock lock(String lockKey, TimeUnit unit, long timeout) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(timeout, unit);
        return lock;
    }

    /**
     * setNX操作
     * @param key  参数
     * @param time 参数
     * @return 返回值
     */
    @Override
    public boolean lockNX(String key, String val, long time, TimeUnit unit) {
        return setIfAbsent(key, val, time, unit);
    }

    /**
     * 尝试加锁-超过最大等待时间则返回
     * @param lockKey
     * @param unit
     * @param waitTime  获取锁等待时间
     * @param leaseTime 释放锁时间，最大持有锁的时间
     * @return
     */
    @Override
    public boolean tryLock(String lockKey, TimeUnit unit, long waitTime, long leaseTime) {
        RLock lock = redissonClient.getLock(lockKey);
        try {
            return lock.tryLock(waitTime, leaseTime, unit);
        } catch (InterruptedException e) {
            System.out.println("加锁异常");
            Thread.currentThread().interrupt();
            return false;
        }
    }

    /**
     * 尝试加锁-连锁机制，所有key成功加锁才算成功
     * @param lockKeys
     * @param unit
     * @param waitTime  获取锁等待时间
     * @param leaseTime 释放锁时间
     * @return
     */
    @Override
    public boolean multiLockTryLock(List<String> lockKeys, TimeUnit unit, long waitTime, long leaseTime) {
        RedissonMultiLock rm = new RedissonMultiLock(lockKeys.stream().map(it -> redissonClient.getLock(it)).toArray(RLock[]::new));
        try {
            return rm.tryLock(waitTime, leaseTime, unit);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }

    @Override
    public void multiLockUnlock(List<String> lockKeys) {
        RedissonMultiLock rm = new RedissonMultiLock(lockKeys.stream().map(it -> redissonClient.getLock(it)).toArray(RLock[]::new));
        unlockSilently(rm);
    }

    /**
     * 释放锁
     *
     * @param lockKey
     */
    @Override
    public void unlock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        if (lock.isHeldByCurrentThread()) {
            unlockSilently(lock);
        }
    }



    private void unlockSilently(RLock lock) {
        try {
            lock.unlock();
        } catch (Exception e) {
            log.error("释放锁失败，key：" + lock.getName(), e);
        }
    }

    @Override
    public Long zSize(String key) {
        try {
            return redisTemplate.opsForZSet().size(key);
        } catch (Exception e) {
            log.error("Redis操作异常，无法获取有序集合数量，key:{}", key, e);
            return 0L;
        }
    }


    /**
     * 根据key获取value，并转换成对应的类型，string类型不进行反序列
     * @param key   缓存key
     * @param clazz 返回对象类型
     * @return
     */
    private <T> Object getSerializ(String key, Class<T> clazz) {
        String result = get(key);
        if (!StringUtils.isEmpty(result)) {
            if (String.class.isAssignableFrom(clazz)) {
                return result;
            }
            return JSON.parseObject(result, clazz);
        }
        return result;
    }

    /**
     * 集合
     * @param key   缓存key
     * @param clazz 返回对象类型
     * @return
     */
    private <T> Object getSerializArray(String key, Class<T> clazz) {
        return gets(key, clazz);
    }

    /**
     * 构建缓存
     *
     * @param key1        缓存key1
     * @param expireTime1 缓存key1过期时间
     * @param key2        缓存key2
     * @param expireTime2 缓存key2过期时间
     * @param result      缓存value
     * @param unit        时间单位
     */
    private void buildCache(String key1, long expireTime1, String key2, long expireTime2, Object result, TimeUnit unit) {
        if (result == null) {
            return;
        }
        if (result instanceof String) {
            set(key1, (String) result, expireTime1, unit);
            set(key2, (String) result, expireTime2, unit);
        } else {
            set(key1, result, expireTime1, unit);
            set(key2, result, expireTime2, unit);
        }
    }
}
